package com.rpcfx;

import com.rpcfx.api.RegistryCenterService;
import com.rpcfx.api.RegistryService;
import com.rpcfx.api.ServiceProviderDesc;
import com.rpcfx.transports.NetUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class DefaultRegistryCenterServiceImpl implements RegistryCenterService {

    private CuratorFramework client;

    private int exportPort = 20881;

    public static final String REGISTRY = "/registry";


    public static final String ROOT = REGISTRY + "/providers";

    public static final String SEPARATE_LINE = "/";

    public String connectingString;

    private static final ConcurrentHashMap<String, Class<?>> REGISTR_MAP = new ConcurrentHashMap<>();

    private static final ConcurrentHashMap<String, List<String>> PROVIDER_MAP = new ConcurrentHashMap<>();


    public DefaultRegistryCenterServiceImpl(String connectingString, int exportPort) {
        try {
            init(connectingString, exportPort);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public DefaultRegistryCenterServiceImpl() {
        this("127.0.0.1:2181", 20881);
    }

    public void init(String connectingString, int exportPort) throws Exception {
        this.connectingString = connectingString;
        this.exportPort = exportPort;
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.builder()
                .connectString(connectingString)
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(50000)
                .retryPolicy(retryPolicy)
                .namespace("RPC")
                .build();
        client.start();
        Stat stat = client.checkExists().forPath(REGISTRY);
        if (stat == null) {
            client.create().withMode(CreateMode.PERSISTENT).forPath(REGISTRY);
        }
        stat = client.checkExists().forPath(ROOT);
        if (stat == null) {
            client.create().withMode(CreateMode.PERSISTENT).forPath(ROOT);
        }
    }


    @Override
    public boolean registry(Class<?> clzz) {
        RegistryService annotation = clzz.getAnnotation(RegistryService.class);
        if (annotation == null) {
            return false;
        }
        String ip = NetUtils.getLocalInetAddress().getHostName();

        ServiceProviderDesc serviceProviderDesc = ServiceProviderDesc.builder().serviceClass(annotation.name()).host(ip).port(exportPort).build();

        try {


            String host = ROOT + serviceProviderDesc.getHost();
            registry(host, CreateMode.PERSISTENT);
            // /registry/providerszhuqidembp-2:20881/com.test.api.UserService
            String service = ROOT + serviceProviderDesc.getServiceName();
            registry(service, CreateMode.EPHEMERAL);

            REGISTR_MAP.put(annotation.name(), clzz);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;

    }

    private void registry(String nodeName, CreateMode createMode) throws Exception {
        Stat stat = client.checkExists().forPath(nodeName);
        if (stat == null) {
            client.create().withMode(createMode).forPath(nodeName);
        }
    }

    @Override
    public Map<String, List<String>> getAllServiceProvider() {

        if (!PROVIDER_MAP.isEmpty()) {
            return PROVIDER_MAP;
        }
        try {
            Map<String, List<String>> tmpServiceMap = new HashMap<>();
            List<String> stringList = client.getChildren().forPath(ROOT);
            for (String host : stringList) {// host:port
                String watchKey = ROOT + SEPARATE_LINE + host;
                client.getChildren().watched().forPath(watchKey);
                List<String> serviceNameList = client.getChildren().forPath(watchKey);//host:port/serviceName
                watchChild(ROOT + SEPARATE_LINE + host);
                for (String serviceName : serviceNameList) {
                    List<String> compute = tmpServiceMap.compute(serviceName, (key, value) -> new ArrayList<>());
                    compute.add(host);

                }
            }
            synchronized (PROVIDER_MAP) {
                PROVIDER_MAP.clear();
                PROVIDER_MAP.putAll(tmpServiceMap);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return PROVIDER_MAP;
    }

    public void watchChild(String nodePath) throws Exception {

        PathChildrenCache pathChildrenCache = new PathChildrenCache(client, nodePath, true);

        //添加监听
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();
                if (type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED) || type.equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
                    System.out.println(pathChildrenCacheEvent);
                    List<String> serviceNameList = client.getChildren().forPath(nodePath);//host:port/serviceName
                    Map<String, List<String>> tmpServiceMap = new HashMap<>();
                    for (String serviceName : serviceNameList) {
                        serviceName = serviceName.substring(1);
                        List<String> compute = tmpServiceMap.compute(serviceName, (key, value) -> new ArrayList<>());
                        compute.add(nodePath.substring(nodePath.lastIndexOf("/") + 1));
                    }
                    PROVIDER_MAP.putAll(tmpServiceMap);
                }
            }
        });
        //开启监听器
        pathChildrenCache.start();
        System.out.println("监听器已开启！");
    }

    @Override
    public Class<?> getClass(String className) {
        return REGISTR_MAP.get(className);

    }
}
